package com.ruoyi.system.mqtt;

import com.ruoyi.common.utils.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;

/**
 * @program: cloudpark
 * @description:
 * @author: cloud
 * @create: 2019-06-27 18:51
 **/
@Slf4j
@Component
public class MqttParkClient {

	private static MqttClient mqttClient = null;

	private static MqttConnectOptions mqttConnectOptions = null;

	public static String MQTT_CLIENTID = "CB_987654321btest1";
//	public static String MQTT_CLIENTID = "CB_987654321bpro";

	public static String MQTT_SERVER = "cloud.tdzntech.com";
//	public static String MQTT_SERVER = "47.104.141.149";

	public static String MQTT_PORT = "1883";

	public static String PARK_SUB_TOPIC = "tpc/TDZN/100760P5575";
//	public static String[] PARK_SUB_TOPIC = {"tpc/TDZN/100760P5575","tpc/TDZN/100760P5596"};

	private SpringContextUtil springContextUtil = new SpringContextUtil();


	/**
	 * 连接MQTT
	 */
	public static void CreateMqttConnection(final String server, final String port, final String clientID,
			final String userName, final String passWord) {
		log.info("MQPark##clientID:{},MQ正在启动...", clientID);
		connectionAndSubscribe(server, port, clientID, userName, passWord);
		log.info("MQPark##clientID:{},MQ初始化完成.", clientID);
	}

	public static void connectionAndSubscribe(String server, String port, String clientID, String userName,
			String passWord) {
		MqttClient mqttClient = getMqttClient(server, port, clientID);
		MqttConnectOptions mqttConnectOptions = getMqttConnectOptions(userName, passWord);

		while (!mqttClient.isConnected()) {
			try {
				mqttClient.connect(mqttConnectOptions);
				log.info("MQPark##当前clientID:{}", mqttClient.getClientId());
			} catch (Exception e) {
				try {
					log.error("MQPark##连接订阅异常,60秒后重新连接", e);
					Thread.sleep(60000);
				} catch (Exception e1) {
					throw new RuntimeException("MQPark##连接订阅异常");
				}
			}
		}
	}

	public static MqttConnectOptions getMqttConnectOptions(String userName, String passWord) {
		if (mqttConnectOptions == null) {
			mqttConnectOptions = new MqttConnectOptions();
			// 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
			mqttConnectOptions.setCleanSession(true);
			mqttConnectOptions.setUserName(userName);
			mqttConnectOptions.setPassword(passWord.toCharArray());
			// 允许同时发送几条消息（未收到broker确认信息）
			mqttConnectOptions.setMaxInflight(1024);
			// 设置超时时间 单位为秒
			mqttConnectOptions.setConnectionTimeout(10);
			// 设置如果连接丢失，客户端是否会自动尝试重新连接到服务器。
			mqttConnectOptions.setAutomaticReconnect(true);
			// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
//			mqttConnectOptions.setKeepAliveInterval(20);
			mqttConnectOptions.setKeepAliveInterval(200);
			// 选择MQTT版本
			mqttConnectOptions.setMqttVersion(MqttConnectOptions.MQTT_VERSION_3_1_1);
		}
		return mqttConnectOptions;
	}

	public static MqttClient getMqttClient(String server, String port, String clientID) {
		try {
			if (mqttClient == null) {
				mqttClient = new MqttClient("tcp://" + server + ":" + port, clientID, new MemoryPersistence());
				mqttClient.setCallback(new MqttParkMessageCallBack());
			}
		} catch (MqttException e) {
			log.error("MQPark##初始化客户端失败", e);
		}
		return mqttClient;
	}

	public static boolean publish(String message, int qos, String topic) {
		try {

			MqttClient mqttClient = MqttParkClient.getMqttClient(MQTT_SERVER, MQTT_PORT, MQTT_CLIENTID);

			log.info("MQPark##发布消息,topic:{},msg:{} ", topic, message);

			MqttMessage _m = new MqttMessage(message.getBytes("utf8"));
			_m.setQos(qos);
			MqttDeliveryToken token = mqttClient.getTopic(topic).publish(_m);

			return token.isComplete(); // 只代表操作完成，不代表发送成功。
										// 在成功完成操作和失败操作的情况下，都将返回True

		} catch (MqttPersistenceException e) {
			log.error("MQPark##发布消息异常 MqttPersistenceException", e);

		} catch (MqttException e) {
			log.error("MQPark##发布消息异常 MqttException", e);

		} catch (UnsupportedEncodingException e) {
			log.error("MQPark##发布消息异常 UnsupportedEncodingException", e);

		} catch (Exception e) {
			log.error("MQPark##发布消息异 Exception", e);
		}
		return false;
	}

	/**
	 * 默发布消息qos=1
	 *
	 * @param message
	 * @return
	 */
	public static boolean publish(String message, String topic) {
		return publish(message, 1, topic);
	}



}
